package com.example.dayu521.msg_sender.conf;

import com.fasterxml.jackson.databind.ObjectMapper;
import org.example.dayu521.contants.AmqpC;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.batch.SimpleBatchingStrategy;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.BatchingRabbitTemplate;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.amqp.support.converter.MessageConverter;
import org.springframework.boot.autoconfigure.amqp.RabbitTemplateConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.scheduling.TaskScheduler;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.web.service.annotation.DeleteExchange;

import java.util.Map;

//springboot
//https://docs.spring.io/spring-boot/docs/current/reference/html/messaging.html#messaging.amqp
@Configuration
public class AmqpConf {
    /**
     * 发布者确认
     *      https://www.rabbitmq.com/confirms.html
     * https://docs.spring.io/spring-amqp/docs/current/reference/html/#cf-pub-conf-ret
     * 配置文件:
     * #配置CachingConnectionFactory,开启发布者确认
     * spring.rabbitmq.publisher-confirm-type=correlated
     * #配置CachingConnectionFactory,开启发布者返回
     * #spring.rabbitmq.publisher-returns=true
     * 配置文件:
     * spring.rabbitmq.listener.simple.prefetch=
     */

    //https://docs.spring.io/spring-amqp/docs/current/reference/html/#async-annotation-conversion
    //https://docs.spring.io/spring-boot/docs/current/reference/html/messaging.html#messaging.amqp.sending
    //MappingJackson2HttpMessageConverterConfiguration用的这种方法配置objectMapper
    //如果不配置,会导Date类型不能转换
    @Bean
    public MessageConverter MyMessageConverter(ObjectMapper objectMapper){
        var c=new Jackson2JsonMessageConverter(objectMapper);
        return c;
    }

    //https://docs.spring.io/spring-amqp/docs/current/reference/html/#Jackson2JsonMessageConverter-to-message
    @Bean
    public RabbitTemplate rabbitTemplate(RabbitTemplateConfigurer rabbitTemplateConfigurer,
                                         ConnectionFactory connectionFactory,
                                         RabbitTemplate.ConfirmCallback confirmCallback,
                                         //springboot自动配置ThreadPoolTaskScheduler
                                         //  https://docs.spring.io/spring-boot/docs/current/reference/html/features.html#features.task-execution-and-scheduling
                                         ThreadPoolTaskScheduler taskScheduler
    ){
//        var t=new RabbitTemplate();
        //批量发送,第三个timeout参数表示如果消息数量没有达到批量大小,那么最多等待多少毫秒就不等待了,直接发送目前已有的消息
        //提交的批量消息放到执行器里执行
        //  https://docs.spring.io/spring-amqp/docs/current/reference/html/#template-batching
        var t = new BatchingRabbitTemplate(new SimpleBatchingStrategy(10,100,100), taskScheduler);
        rabbitTemplateConfigurer.configure(t, connectionFactory);
//        t.setReturnsCallback(rv->{
//            System.out.println("消息不能路由到指定队列,请检查biding是否正确");
//        });
        //https://docs.spring.io/spring-amqp/docs/current/reference/html/#template-confirms
        //消息到达交换机时调用.如果交换机不存在,ack参数为false,否则为true
//        t.setConfirmCallback(confirmCallback);
        //消息能被路由到队列,就会回调这个接口,
//        t.setMandatory(true);
//        t.setReturnsCallback();

        //https://docs.spring.io/spring-amqp/docs/current/reference/html/#post-processing
        //设置消息持久化
//        t.addBeforePublishPostProcessors(message -> {
//            //默认就是持久化的
//            message.getMessageProperties().setDeliveryMode(MessageDeliveryMode.PERSISTENT);
//            return message;
//        });
        return t;
    }

    @Bean
    @Primary
    public RabbitTemplate.ConfirmCallback myConfirmCallback(){
        return (CorrelationData correlation, boolean ack, String s)->{
            if (correlation != null) {
                System.out.println("Received " + (ack ? " ack " : " nack ") + "for correlation: " + correlation);
            }
        };
    }

    @Bean
    public Queue myqueue(){
        return QueueBuilder
                .durable(AmqpC.Shangjia.msg_sender_queue)
//                .maxPriority(10) //设置队列最大优先级,然后发消息时使用messageproperty来设置消息的优先级
//                .lazy() //惰性队列.不把消息放到内存中,会放到磁盘.内存中放索引,所以消费速度慢
                .build();
        //队列持久化
//        return new Queue(AmqpC.Shangjia.msg_sender_queue,true,false,false);
    }

    //返回类型需要是具体的DirectExchange,而不能是Exchange接口,因为下面创建binding时不接受Exchange
    @Bean
    public DirectExchange helloExchange() {
        return ExchangeBuilder.directExchange(AmqpC.Shangjia.msg_sender_exchange).durable(true).build();
        //交换机持久化
//        return new DirectExchange(AmqpC.Shangjia.msg_sender_exchange,true,false);
    }

    @Bean
    public Binding binding(){
        return BindingBuilder.bind(myqueue()).to(helloExchange()).with(AmqpC.Shangjia.msg_sender_key);
//        return new Binding(AmqpC.Shangjia.msg_sender_queue
//                ,Binding.DestinationType.QUEUE
//                ,AmqpC.Shangjia.msg_sender_exchange
//        ,AmqpC.Shangjia.msg_sender_key, Map.of());
    }

}
